package com.chris.core.socket.netty.util;

import com.chris.core.socket.netty.code.CustomMessageDecoder;
import com.chris.core.socket.netty.code.CustomMessageEncoder;
import com.chris.core.socket.netty.dto.SocketTypeEnum;
import com.chris.core.socket.netty.handler.heatbeat.HeatBeatCustomHandler;
import com.chris.core.socket.netty.handler.heatbeat.HeatBeatStringHandler;

import com.chris.core.socket.netty.handler.heatbeat.HeatBeatObjectHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;

import io.netty.handler.codec.bytes.ByteArrayDecoder;
import io.netty.handler.codec.bytes.ByteArrayEncoder;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.handler.timeout.ReadTimeoutHandler;

import io.netty.util.CharsetUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.net.InetSocketAddress;
import java.util.concurrent.TimeUnit;


/**
 * netty 通用类 客户端
 * 2023-5-15重新检视
 */
public abstract class NettyClientUtil {

    //#region 参数及变量

    private Logger log = LoggerFactory.getLogger(this.getClass());

    private String host;

    /**
     * 端口
     */
    private int port;

    /**
     * 服务器读取超时时间（秒）
     */
    private int readTimeOut = 180;

    /**
     * 消息断言
     */
    public String breakMark = "\r\n";// \n    \r\n    ~!@#bbq8$%^&*\n;

    /**
     * 服务器类型
     * “” 普通类型  |  websocket
     */
    private SocketTypeEnum serverType = SocketTypeEnum.ByteSocket;


    /**
     * 当失去连接时，是否自动
     */
    private boolean isAutoConnected = true;

    /**
     * 客户端事件循环线程组
     * zhu
     */
    static EventLoopGroup group;//= new NioEventLoopGroup();

    /**
     * 客户端启动对象
     */
    static Bootstrap bootstarp;// = new Bootstrap();

    ChannelFuture channelFuture;


    //#endregion


    //#region 构造函数及其初始化


    /**
     * 构造函数
     *
     * @param port            端口
     * @param readTimeOut     读取超时
     * @param breakMark       断言
     * @param isAutoConnected 自动连接
     * @param serverType      服务器类型
     */
    public NettyClientUtil(String host,int port, int readTimeOut, String breakMark, boolean isAutoConnected, SocketTypeEnum serverType) {
        this.host=host;
        this.port = port;
        this.breakMark = breakMark;
        this.serverType = serverType;
        this.readTimeOut = readTimeOut;
        this.isAutoConnected = isAutoConnected;
    }


    /**
     * 初始化系统配置
     */
    private void IntSystem() {

        try {
            //----------------注意-----------------------------
            //此处有坑，要注意，每个 NioEventLoopGroup 会占用30倍所有的tcp有效连接，
            // 压力测试时候，如果上到2000个客户端实例（每个压力测试机只能创建1500个客户端）， 60000个有效连接就满了
            //group ,  bootstarp 都变成全局静态 （相对客户端而已）
            if (group != null && bootstarp != null)
                return;

            group = new NioEventLoopGroup();
            bootstarp = new Bootstrap();
            bootstarp.group(group)//设置线程组
                    .channel(NioSocketChannel.class)//设置客户端通道类型
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();

                            if (serverType == SocketTypeEnum.ByteSocket) {
                                //#region -------------编解码---------------
                                //断言拆包pipline必须放在第一位，内容先做buff截留，拆包后才把内容out.add抛给下家decoder
                                if (breakMark != "") {
                                    ByteBuf delimiter = Unpooled.wrappedBuffer(breakMark.getBytes(CharsetUtil.UTF_8));
                                    pipeline.addLast(new DelimiterBasedFrameDecoder(Integer.MAX_VALUE,
                                            delimiter));//, delimiter_1
                                }
                                //解码器
                                pipeline.addLast("decoder", new ByteArrayDecoder());
                                //编码器
                                pipeline.addLast("encoder", new ByteArrayEncoder());
                                //#endregion

                                //#region -------------心跳---------------
                                //心跳策略：
                                // (1)客户端  检测写心跳 , 30s 没有内容自动发送ping , 不断开 , 服务端不回复 pong  ; 检测读 60s *3 心跳 , 没有内容动自动断开连接
                                // (2)服务端  检测写心跳 , 60s 没有内容自动发送ping , 不断开 , 客户端不回复 pong  ; 检测读 30s *6 心跳 , 没有内容动自动断开连接

                                //设置时间段内都没有数据读取了，那么就引发超时，然后关闭当前的channel --> 会自动断开
                                //pipeline.addLast(new ReadTimeoutHandler(readTimeOut));
                                //控制数据输出的时候的超时，如果在设置时间段内都没有数据写了，那么就超时--> 会自动断开
                                //pipeline.addLast(new WriteTimeoutHandler(30*60));
                                /**
                                 * 空闲状态处理器
                                 * readerIdleTime : 空闲 ？ 读取时间 ， 自动发送心跳检测包
                                 * writerIdleTime ：空闲 ？ 写时间， 自动发送心跳包
                                 * allIdleTime： 空闲 ？ 读+写时间，自动发送心跳包
                                 */
                                //pipeline.addLast(new IdleStateHandler(0, 60, 0, TimeUnit.SECONDS));
                                //重要：上下handler会互相关联，加入自定义心跳事件，对应上面函数IdleStateHandler-->会触发下面的handler-->HeatBeatHandler
                                //pipeline.addLast(new HeatBeatHandler());

                                //-------------------------------------------
                                //#endregion
                            } else if (serverType == SocketTypeEnum.StringSocket) {

                                //#region -------------编解码---------------
                                //断言拆包pipline必须放在第一位，内容先做buff截留，拆包后才把内容out.add抛给下家decoder
                                if (breakMark != "") {
                                    //同时存在多个分隔符时，优先匹配长度最短的分隔符，如果一样长，则哪个先出现，匹配哪个。
                                    ByteBuf delimiter = Unpooled.wrappedBuffer(breakMark.getBytes(CharsetUtil.UTF_8));

                                    //ByteBuf delimiter_1 = Unpooled.wrappedBuffer("\n".getBytes(CharsetUtil.UTF_8));
                                    // 将delimiter设置到DelimiterBasedFrameDecoder中，经过该解码一器进行处理之后，源数据将会
                                    // 被按照_$进行分隔，这里1024指的是分隔的最大长度，即当读取到1024个字节的数据之后，若还是未
                                    // 读取到分隔符，则舍弃当前数据段，因为其很有可能是由于码流紊乱造成的
                                    pipeline.addLast(new DelimiterBasedFrameDecoder(Integer.MAX_VALUE,delimiter));
                                }

                                //编码器 文本转换为utf-8
                                pipeline.addLast("decoder", new StringDecoder(CharsetUtil.UTF_8));
                                //编码器
                                pipeline.addLast("encoder", new StringEncoder(CharsetUtil.UTF_8));
                                //#endregion

                                //#region -----------心跳------------
                                //心跳策略：
                                // (1)客户端  检测写心跳 , 30s 没有内容自动发送ping , 不断开 , 服务端不回复 pong  ; 检测读 60s *3 心跳 , 没有内容动自动断开连接
                                // (2)服务端  检测写心跳 , 60s 没有内容自动发送ping , 不断开 , 客户端不回复 pong  ; 检测读 30s *6 心跳 , 没有内容动自动断开连接


                                //设置时间段内都没有数据读取了，那么就引发超时，然后关闭当前的channel --> 会自动断开
                                pipeline.addLast(new ReadTimeoutHandler(readTimeOut));
                                //控制数据输出的时候的超时，如果在设置时间段内都没有数据写了，那么就超时--> 会自动断开
                                // pipeline.addLast(new WriteTimeoutHandler(30*60));
                                /**
                                 * 空闲状态处理器
                                 * readerIdleTime : 空闲 ？ 读取时间 ， 自动发送心跳检测包
                                 * writerIdleTime ：空闲 ？ 写时间， 自动发送心跳包
                                 * allIdleTime： 空闲 ？ 读+写时间，自动发送心跳包
                                 */
                                pipeline.addLast(new IdleStateHandler(0, 30, 0, TimeUnit.SECONDS));
                                //重要：上下handler会互相关联，加入自定义心跳事件，对应上面函数IdleStateHandler-->会触发下面的handler-->HeatBeatHandler
                                pipeline.addLast(new HeatBeatStringHandler());
                                //#endregion
                                //--------------------
                            } else if (serverType == SocketTypeEnum.ObjectSocket) {
                                //#region -------------编解码---------------
                                //解码器 decoder 自带拆包功能
                                pipeline.addLast("decoder", new ObjectDecoder(Integer.MAX_VALUE, ClassResolvers.cacheDisabled(null)));

                                //编码器
                                pipeline.addLast("encoder", new ObjectEncoder());

                                //#endregion
                                //#region -------------心跳---------------

                                // (1)客户端  检测写心跳 , 30s 没有内容自动发送ping , 不断开 , 服务端不回复 pong  ; 检测读 60s *3 心跳 , 没有内容动自动断开连接
                                // (2)服务端  检测写心跳 , 60s 没有内容自动发送ping , 不断开 , 客户端不回复 pong  ; 检测读 30s *6 心跳 , 没有内容动自动断开连接

                                //设置时间段内都没有数据读取了，那么就引发超时，然后关闭当前的channel --> 会自动断开
                                pipeline.addLast(new ReadTimeoutHandler(readTimeOut));
                                //控制数据输出的时候的超时，如果在设置时间段内都没有数据写了，那么就超时--> 会自动断开
                                //pipeline.addLast(new WriteTimeoutHandler(30 * 5));
                                /**
                                 * 空闲状态处理器
                                 * readerIdleTime : 空闲 ？ 读取时间 ， 自动发送心跳检测包
                                 * writerIdleTime ：空闲 ？ 写时间， 自动发送心跳包
                                 * allIdleTime： 空闲 ？ 读+写时间，自动发送心跳包
                                 */
                                pipeline.addLast(new IdleStateHandler(0, 30, 0, TimeUnit.SECONDS));
                                //重要：上下handler会互相关联，加入自定义心跳事件，对应上面函数IdleStateHandler-->会触发下面的handler-->HeatBeatHandler
                                pipeline.addLast(new HeatBeatObjectHandler());
                                //#endregion

                                //----------------------------------------


                            }
                            else if (serverType == SocketTypeEnum.CustomSocket) {
                                //#region -------------编解码---------------

                                //解码器 decoder 自带拆包功能
                                pipeline.addLast("decoder", new CustomMessageDecoder());

                                //编码器
                                pipeline.addLast("encoder", new CustomMessageEncoder());

                                //#endregion

                                //#region -------------心跳---------------

                                // (1)客户端  检测写心跳 , 30s 没有内容自动发送ping , 不断开 , 服务端不回复 pong  ; 检测读 60s *3 心跳 , 没有内容动自动断开连接
                                // (2)服务端  检测写心跳 , 60s 没有内容自动发送ping , 不断开 , 客户端不回复 pong  ; 检测读 30s *6 心跳 , 没有内容动自动断开连接

                                //设置时间段内都没有数据读取了，那么就引发超时，然后关闭当前的channel --> 会自动断开
                                pipeline.addLast(new ReadTimeoutHandler(readTimeOut));
                                //控制数据输出的时候的超时，如果在设置时间段内都没有数据写了，那么就超时--> 会自动断开
                                //pipeline.addLast(new WriteTimeoutHandler(30 * 5));
                                /**
                                 * 空闲状态处理器
                                 * readerIdleTime : 空闲 ？ 读取时间 ， 自动发送心跳检测包
                                 * writerIdleTime ：空闲 ？ 写时间， 自动发送心跳包
                                 * allIdleTime： 空闲 ？ 读+写时间，自动发送心跳包
                                 */
                                pipeline.addLast(new IdleStateHandler(0, 30, 0, TimeUnit.SECONDS));
                                //重要：上下handler会互相关联，加入自定义心跳事件，对应上面函数IdleStateHandler-->会触发下面的handler-->HeatBeatHandler
                                pipeline.addLast(new HeatBeatCustomHandler());
                                //#endregion

                                //----------------------------------------


                            }
                            AddChannelHandler(pipeline);
                        }
                    });

        } catch (Exception ex) {
            if (group != null)
                group.shutdownGracefully();
        } finally {
            //  group.shutdownGracefully();
        }

    }

    /**
     * 添加自定义 Handler
     * 可以重载， 并加上 super.AddChannelHandler(pipeline);
     */
    public abstract void AddChannelHandler(ChannelPipeline pipeline);


    /**
     * 打开端口
     *
     * @throws InterruptedException
     */
    public void Open() {
        Thread t = new Thread(new Runnable() {
            @Override
            public void run() {
                IntSystem();
                Connect();
            }

        });
        t.start();

    }

    /**
     * 客户端连接事件
     */
    private void Connect() {
        try {
            Thread.sleep(3000);
            InetSocketAddress address = new InetSocketAddress(host,port);
            //启动监听
            // channelFuture = bootstarp.connect(address).sync();
            //同步方式执行
            channelFuture = bootstarp.connect(address).sync();
            //log.info("Netty Socket 客户端 启动  端口:" + port);
            //启动关闭通道监听
            channelFuture.channel().closeFuture().sync();

        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {

            if (group != null)
                group.shutdownGracefully();

            if (channelFuture != null)
                channelFuture = null;

            if (isAutoConnected) {
                Open();
            }
        }
    }

    /**
     * 停止服务
     */
    private void Stop() {
        if (group != null)
            group.shutdownGracefully();
        if (channelFuture != null)
            channelFuture = null;
    }
    //#endregion


    //#region 发送消息

    /**
     * 发送消息到服务端
     */
    public void sendStrMsg(String msg) {
        if (channelFuture != null && channelFuture.isSuccess()) {
            Channel channel = channelFuture.channel();
            NettyRunTimeUtil.sendStrMsg(msg,channel,breakMark);
        }
    }


    /**
     * 发送消息到服务端
     */
    public void sendObjMsg(Object ob) {
        if (channelFuture != null && channelFuture.isSuccess()) {
            Channel channel = channelFuture.channel();
            NettyRunTimeUtil.sendObjMsg(ob,channel);
        }
    }



    /**
     * 发送文件命令
     * @param file
     * @throws Exception
     */
    @Deprecated
    public void sendFileMsg(File file) throws Exception {
        if (channelFuture != null && channelFuture.isSuccess()) {
            Channel channel = channelFuture.channel();
            NettyRunTimeUtil.sendFileMsg(file,channel);
        }
    }

    /**
     * 发送文件（object->CustomMessageProtocol)
     * @param file
     * @throws Exception
     */
    public void sendCustomFileMsg(File file) throws Exception {
        if (channelFuture != null && channelFuture.isSuccess()) {
            Channel channel = channelFuture.channel();
            NettyRunTimeUtil.sendCustomFileMsg(file,channel,log);
        }
    }

    /**
     * 发送消息到服务端
     */
    public void sendByteMsg(byte[] bytes) {
        if (channelFuture != null && channelFuture.isSuccess()) {
            Channel channel = channelFuture.channel();
            NettyRunTimeUtil.sendByteMsg(bytes,channel,breakMark);
        }
    }



    //#endregion


    //#region 流量控制

    /**
     * 流量控制  (目前策略 接收是 发送 2 被 限流 ，双方需要约定好 ）
     * 这是接收端端限制，因为是并发异步/同步，所以不能同步阻止发送端写速（NIO/BIO）
     * 只能总限制 并发总数 * 单并发 = 总速度 , 例如 ： 限制 200KB * 100 = 20 M / S  ,1000并发 200M， 1W并发 2000M
     *
     * @param passedlen  总已传输字节 (b！,非KB）
     * @param beginTime  开始时间
     * @param limitspped 限制速度
     */
    private static void flowControl(long passedlen, long beginTime, long limitspped) {
        //流量控制 100KB/S = 100  * 1024  b/s  = 1024000 b/S =  102 b /ms
        //limitspped KB /s = limitspped * 1024 * 1000 / 1000 = limitspped * 1024  / ms
        long passTime = (System.currentTimeMillis() - beginTime);
        if (passTime == 0) passTime = 1;
        long pp = passedlen / limitspped - passTime;//如果passTime=0，按实际所需passedlen/limitspped等待
        if (pp > 0) {
            try {
                Thread.sleep(pp);

            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    //#endregion
}
